package com.voice.client.channel.tcp;

import com.voice.client.channel.DefaultStreamListener;
import com.voice.client.channel.handle.impl.DeviceChannelStreamListener;
import com.voice.client.message.impl.MessageClientImpl;
import com.voice.core.channel.tcp.impl.IOSelectorConnectionManager;
import com.voice.core.channel.tcp.impl.IOSelectorContext;
import com.voice.core.connection.Connection;
import com.voice.core.connection.ConnectionManager;
import com.voice.core.utils.CloseUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class TCPServer implements Closeable{

    private DefaultStreamListener defaultStreamListener;
    private ConnectionManager connectionManager;
    private int port;
    private AtomicBoolean closed;
    private ServerSocketChannel server;
    private Selector selector;
    private TCPListen tcpListen;


    public TCPServer(int port, MessageClientImpl messageClient) throws IOException {
        this.port = port;
        this.defaultStreamListener = new DefaultStreamListener(messageClient,new DeviceChannelStreamListener(messageClient));
        this.connectionManager = new IOSelectorConnectionManager(new IOSelectorContext());
        this.closed=new AtomicBoolean(true);
    }

    public void start() throws IOException {
        try {
            selector = Selector.open();
            closed.set(false);
            server = ServerSocketChannel.open();
            server.configureBlocking(false);
            server.bind(new InetSocketAddress(port));
            server.register(selector, SelectionKey.OP_ACCEPT);
            tcpListen = new TCPListen();
            tcpListen.start();
        } catch (IOException e) {
            throw e;
        }
    }

    @Override
    public void close() {
        if(closed.compareAndSet(false,true)){
            selector.wakeup();
            connectionManager.shutdown();
            CloseUtil.close(server, selector);
        }
    }

    class TCPListen extends Thread {
        @Override
        public void run() {
            log.info("服务器已经启动,正在监听端口号:{}", port);
            while (!closed.get()) {
                try {
                    if (selector.select() == 0) continue;
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    for (SelectionKey selectionKey : selectionKeys) {
                        ServerSocketChannel server = (ServerSocketChannel) selectionKey.channel();
                        SocketChannel socketChannel = server.accept();
                        Connection connection = connectionManager.getConnection(socketChannel);
                        connection.setDefaultStreamListener(defaultStreamListener);
                    }
                    selectionKeys.clear();
                } catch (IOException e) {
                    log.error("客户端连接异常:{}", e);
                }
            }
            log.info("服务器已经关闭");
        }
    }

}
